package com.gis.aoitest.processfunction;

import com.gis.aoitest.beans.SensorReading;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author LnnuUser
 * @create 2021-09-09-上午8:20
 */
public class ProcessTest01_KeyedProcessFunction {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // socket文本流
        DataStreamSource<String> inputStream = env.socketTextStream("localhost", 9999);

        // 转换成SensorReading类型
        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        } );

        // 测试KeyedProcessFunction，先分组然后自定义处理
        SingleOutputStreamOperator<Integer> processStream = dataStream.keyBy("id")
                .process(new MyProcess());

        processStream.print();

        env.execute();
    }

    // 实现自定义的处理函数
    public static class MyProcess extends KeyedProcessFunction<Tuple, SensorReading, Integer> {

        ValueState<Long> tsTimer;

        @Override
        public void open(Configuration parameters) throws Exception {
            tsTimer = getRuntimeContext().getState(new ValueStateDescriptor<Long>("tsTime", Long.class));
        }

        @Override
        public void processElement(SensorReading value, Context context, Collector<Integer> out) throws Exception {

            out.collect(value.getId().length());

            // context
            context.timestamp();
            context.getCurrentKey();
//            context.output();
            // 当前系统时间
            context.timerService().currentProcessingTime();
            // 当前的水位线
            context.timerService().currentWatermark();
            // 注册事件定时器，定时器时间：当前系统时间+1s
            context.timerService().registerEventTimeTimer(context.timerService().currentProcessingTime() + 1000L);
            // 定时器更新为：当前系统时间+1s
            tsTimer.update(context.timerService().currentProcessingTime() + 1000L);
            // 注册系统定时器：事件事件+10s
            context.timerService().registerProcessingTimeTimer((value.getTimestamp() + 10) * 1000);
            // 删除事件定时器，删除时机：当前系统事件+1s
            context.timerService().deleteEventTimeTimer(context.timerService().currentProcessingTime() + 1000L);
            // 删除系统定时器，删除时机：当前定时器时间 -> 系统时间+1s
            context.timerService().deleteProcessingTimeTimer(tsTimer.value());
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Integer> out) throws Exception {
            System.out.println(timestamp + "定时器触发");
        }
    }
}
